/*
* @file_name: event_poller.h
* @date: 2021/10/16
* @author: oaho
* Copyright @ hz oaho, All rights reserved.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef OAHO_EVENT_POLLER_H
#define OAHO_EVENT_POLLER_H
#include <list>
#include <mutex>
#include <functional>
#include <memory>
#include <thread>
#include <atomic>
#include <cstdio>
#include <vector>
#include <map>
#include <chrono>
#include <iostream>
#ifdef __linux__
#include <sys/epoll.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/eventfd.h>
#include "event.h"
#include "file_descriptor.h"
#include "Pipe.h"
extern int errno;
/*
* https://man7.org/linux/man-pages/man7/epoll.7.html
* epoll api performs a similar task to poll(2);monitoring multiple file descriptors
* to see if I/O is possible on any of them, The epoll API can be used either as an
* edge-triggered or a level-triggered interface and scales well to large numbers of
* watched file descriptors.
*
* Level-triggered and edge-triggered
* The epoll event distribution interface is able to behave both as edge-triggered(ET)
* and level-triggered(LT).
* the difference between the two mechanisms,Suppose that this scenario happens:
*	1. the file discriptor that represents the read side of a pipe(rfd) is registered on
*	the epoll fd.
*   2. A pipe writer writes 2 kB of data on the write side of the pipe.
*	3. A call to epoll_wait(2) is done that will return rfd as a ready file descriptor.
*	4. The pipe reader reads 1 kB of data from rfd.
*   5. A call to epoll_wait(2) is done.
*
* if use EPOLLET flag, done in step 5 will probably hang despite the available data still
* present in the file input buffer;The reason for this is that edge-triggered mode
* delivers events only when changes occur on the monitored file descriptor.
*	a) with nonblocking file descriptors;
*   b) by waiting for an event only after read(2) or write(2) return
*      EAGAIN
*/
static constexpr int default_interval = 10000;
class event_epoller {
public:
    using Ptr = std::shared_ptr<event_epoller>;
public:
    event_epoller();
    ~event_epoller();
public:
    template<typename F, typename...Args>
    void async(F&& f, Args&&...args) {
        auto func = std::move(std::bind(f, std::forward<Args>(args)...));
        {
            std::lock_guard<std::mutex> lmtx(_mtx_);
            task_que.emplace_back(std::move(func));
        }
        notify_on_task_arrived();
    }
    template<typename F, typename...Args>
    void run_after(size_t interval, bool run_once, F&& f, Args...args)
    {
        auto func = std::move(std::bind(f, std::forward<Args>(args)...));
        auto ev_ptr = Event::make_basic_event(true, interval, run_once, std::move(func));
        /* add to timer event */
        /* add to this thread */
        int64_t duration_now = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        /* time_point + offset */
        size_t now = (size_t)duration_now + interval;
        async([this, now, ev_ptr]{
            timer_loop.insert(std::make_pair(now, std::move(ev_ptr)));
        });
    }
    inline void set_exception_handler(std::function<void(int)>&& f);
    void register_event(Event::Ptr);
    void remove_event(int fd); 
    inline void remove_event(Event::Ptr);
    inline void start();
    inline void stop();
    inline const std::thread::id& get_thread_id();
private:
    void init();
    void register_event_l();
    int  get_min_delay();
    template<typename F> inline void async_l(F&& f);
    inline void notify_on_task_arrived();
    inline void handle_error();
    inline void handle_event(const epoll_event&);
    inline void handle_self_event(int fd);
    inline void run();
private:
    /* for epoll */
    int epoll_fd;
    /* thread_id */
    std::thread::id _ctid_;
    /* for put task in loop */
    int event_fd;
    /* for put event fd */
    std::vector<epoll_event> events;
    std::atomic<bool> _stop;
    std::unique_ptr<std::thread> _runner;
    std::mutex _mtx_;
    std::list<std::function<void()>> task_que;
    std::map<int, Event::Ptr> event_loop;
    /* 
    * for timer event
    * @key: miliseconds
    * @val: Event
    */
    std::multimap<size_t, Event::Ptr> timer_loop;
    std::function<void(int)> exception_handler_func;
};

event_epoller::event_epoller() :epoll_fd(-1), event_fd(-1), _stop(true) {
    exception_handler_func = [](int code) {
        //fprintf(stderr, strerror(code));
    };
}
event_epoller::~event_epoller() {
    stop();
}
inline void event_epoller::notify_on_task_arrived() {
    int ret = eventfd_write(event_fd, 1);
    if (ret < 0)handle_error();
}
inline void event_epoller::handle_error() {
    int err_code = errno;
    exception_handler_func(err_code);
}
inline void event_epoller::handle_self_event(int fd) {
    int ret = 0;
    eventfd_t val = 0;
    while((ret = eventfd_read(event_fd, &val)) == 0);
    if (ret < 0) {
        int erro = errno;
        if (errno != EAGAIN || errno != EWOULDBLOCK){
            handle_error();
            return;
        }
    }
    decltype(task_que) tmp_task_que;
    {
        std::lock_guard<std::mutex> lmtx(_mtx_);
        tmp_task_que.swap(task_que);
    }
    while (!tmp_task_que.empty()) {
        auto f = std::move(tmp_task_que.front());
        tmp_task_que.pop_front();
        if (f)f();
    }
}

inline void event_epoller::handle_event(const epoll_event& ev) {
    auto it = event_loop.find(ev.data.fd);
    if (it == event_loop.end()) {
        throw std::runtime_error("current fd descriptor is not in loop");
    }
    auto& evt = *(it -> second);
    /* call back */
    evt();
    if (evt.run_once())remove_event(it->first);
}

inline void event_epoller::set_exception_handler(std::function<void(int)>&& f) {
    if (f)exception_handler_func = std::move(f);
}
/* flag: https://man7.org/linux/man-pages/man2/epoll_ctl.2.html */
void event_epoller::register_event(Event::Ptr p) {
    if (!p)return;
    const auto& ev_ptr = *p;
    int ret = file_descriptor::make_nonblocking(ev_ptr.fd());
    if (ret < 0) {
        handle_error();
        return;
    }
    epoll_event ev;
    ev.data.fd = ev_ptr.fd();
    ev.events =  ev_ptr.flags();
    ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, ev_ptr.fd(), &ev);
    if (ret < 0) {
        handle_error();
        return;
    }
    event_loop[ev_ptr.fd()] = std::move(p);
}

void event_epoller::remove_event(Event::Ptr ptr) {
    if(ptr)return remove_event(ptr->fd());
}

void event_epoller::remove_event(int fd) {
    int ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
    if (ret < 0) {
        handle_error();
        return;
    }
    event_loop.erase(fd);
}
void event_epoller::register_event_l() {
    epoll_event ev;
    /* pre the position in events vector */
    ev.data.u64 = events.size();
    ev.data.fd = event_fd;
    ev.events = EPOLLIN | EPOLLET;
    int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, event_fd, &ev);
    if (ret < 0) {
        handle_error();
        return;
    }
    events.emplace_back(ev);
}

int event_epoller::get_min_delay() {
    /* �õ���ǰ��time_point*/
    auto fetch_now = []()->size_t {
        auto duration_now = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        return (size_t)duration_now;
    };
    decltype(timer_loop) _tmp_timer_loop;
    using iterator = decltype(timer_loop)::iterator;
    _tmp_timer_loop.swap(timer_loop);
    /* �õ���ǰ�����̵�һ������ */
    size_t now = 0;
    for (iterator it = _tmp_timer_loop.begin(); it != _tmp_timer_loop.end();++it) {
        auto& ev = *(it->second);
        now = fetch_now();
        /* ���С�ڵ�ǰ��ʱ��,�����ʱ��� */
        if (now < it->first) {
            timer_loop.insert(it, _tmp_timer_loop.end());
            break;
        }
        /* �����ǰ��ʱ�����first,˵�������Ѿ�����,����ִ�� */
        else if (now >= it->first && ev.read_func) {
            ev.read_func(*(it->second));
            /* ѭ��ִ�е����� */
            if (!ev.run_once()) {
                /* ����ȡ��ʱ�䲢������ */
                timer_loop.insert(std::make_pair(fetch_now() + ev.interval, it->second));
            }
        }
    }
    auto min_delay = timer_loop.empty() ? default_interval : timer_loop.begin() -> first - now;
    return (int)min_delay;
}

void event_epoller::init() {
    epoll_fd = epoll_create(5);
    if (epoll_fd == -1) {
        int err_code = errno;
        throw std::runtime_error(strerror(err_code));
    }
    /* Linux 2.6.27 */
    event_fd = eventfd(0, EFD_NONBLOCK);
    if (event_fd == -1) {
        int err_code = errno;
        throw std::runtime_error(strerror(err_code));
    }
    /* put event_fd in epollfd */
    register_event_l();
}
void event_epoller::start()
{
    if (!_stop)return;
       /* ������Ҫͬ��������ȷ��_runner���׼������*/
      _runner.reset(new std::thread([this]() {
          init();
          run();
      }));
      while (_stop)std::this_thread::sleep_for(std::chrono::microseconds(200));
}
void event_epoller::stop()
{
    //_stop = true;
    if (_runner && _runner->joinable())_runner->join();
    close(epoll_fd);
}
void event_epoller::run()
{
    _ctid_ = std::this_thread::get_id();
    int min_delay = default_interval;
    /* ��ʱ������������� */
    _stop = false;
    while (!_stop) {
        int cnt = epoll_wait(epoll_fd, const_cast<epoll_event*>(events.data()), events.size(), min_delay);
        if (cnt < 0) {
            handle_error();
            continue;
        }
        for (int i = 0; i < cnt; i++) {
            /* event fd */
            if (events[i].data.fd == event_fd && events[i].events & EPOLLIN) {
                handle_self_event(events[i].data.fd);
            }
            else handle_event(events[i]);
        }
        /* execute timer tasks, also update min_delay */
        min_delay = get_min_delay();
    }
}
inline const std::thread::id& event_epoller::get_thread_id(){
    if(_stop){
        throw std::runtime_error("current event_poller not have thread id[not run on this time_point]");
    }
    return _ctid_;
}

template<typename F> 
inline void event_epoller::async_l(F&& f) {

}
#endif
#endif
